package cn.lagou.sparkcore.work

import org.apache.spark.broadcast.Broadcast
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Work1 {
  def main(args: Array[String]): Unit = {
    // 创建SparkContext
    val conf = new SparkConf().setAppName(this.getClass.getCanonicalName.init).setMaster("local[*]")
    val sc = new SparkContext(conf)
    sc.setLogLevel("WARN")
    // 业务逻辑
    //读取文件数据
    val httpLines: RDD[String] = sc.textFile("./src/main/data/http.log")
    val ipLines:RDD[String] = sc.textFile("./src/main/data/ip.dat")
    //提取ip段与区域关系
    val ipCitys: RDD[(String, String,String)] = ipLines.map(line => {
      val arrs: Array[String] = line.split("\\|")
      (arrs(0),arrs(1), arrs(7))
    })
    //将ip与区域关系定为广播变量
    val ipCitysBroadcast: Broadcast[Array[(String, String, String)]] = sc.broadcast(ipCitys.collect())
    //获取http日志中访问IP
    val logIpRDD: RDD[String] = httpLines.map(line => {
      line.split("\\|")(1)
    })
    //获取区域统计数RDD
    val cityCount: RDD[(String, Int)] = logIpRDD.mapPartitions(iter => {
      iter.map(ip => {
        val ipLong: Long = ipToLong(ip)
        val index: Int = binarySearch(ipLong, ipCitysBroadcast.value)
        if(index>=0){
          (ipCitysBroadcast.value(index)._3,1)
        }else{
          ("",0)
        }
      })
    })
    //将区域统计数合并输出
    cityCount.reduceByKey(_+_).collect().foreach(println)
    // 关闭SparkContext
    sc.stop()
  }

  //将IP地址转换为long，便于计算区间
  def ipToLong(ip:String) : Long = {
    val ipSplit: Array[String] = ip.split("\\.")
    var ipLong:Long = 0
    for(ipNum <- ipSplit){
      ipLong = ipNum.toLong | ipLong << 8L
    }
    ipLong
  }

  //使用二分法查找ip落到哪个区域IP区间
  def binarySearch(ipLong:Long,ipCity:Array[(String, String, String)]): Int ={
    var st = 0
    var ed = ipCity.length-1
    while (st <= ed) {
      var middle = (st + ed) / 2
      //如果ip大于初始值，小于结束值，直接返回
      if (ipLong >= ipToLong(ipCity(middle)._1) && ipLong <= ipToLong(ipCity(middle)._2)) {
        return middle
      }
      //如果ip小于中间值，ip位于左边
      if (ipLong < ipToLong(ipCity(middle)._1)) {
        ed = middle - 1
      }
      //如果ip大于中间值，ip位于右边
      if (ipLong > ipToLong(ipCity(middle)._2)) {
        st = middle + 1
      }
    }
    -1
  }
}
